package com.demo.rocketmq.spring.support;

import com.demo.rocketmq.constants.RocketMqConstants;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.MethodParameter;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.converter.SmartMessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.util.List;
import java.util.Objects;

/**
 * RocketMQ Container服务创建Rocket Consumer和监听主题消费消息
 * @author cyj
 */
@Slf4j
@Data
public class RocketMqListenerContainer implements SmartLifecycle, InitializingBean {

    private String nameServer;
    private String consumerGroup;
    private String topic;
    private String callbackTopic;
    private String name;
    private String selectorExpression;
    private String charset = "UTF-8";

    private Object bean;

    private Method method;

    private boolean running;

    /**
     * Message consume retry strategy<br> -1,no retry,put into DLQ directly<br> 0,broker control retry frequency<br>
     * >0,client control retry frequency.
     */
    private int delayLevelWhenNextConsume = 0;

    /**
     * 客户端处理消息线程数
     * RocketMQ默认值是20
     * 并且使用new LinkedBlockingQueue<>() 无界队列，所以consumeThreadMax设置无效
     */
    private int consumeThreadMin = 20;
    private int consumeThreadMax = 20;

    private long suspendCurrentQueueTimeMillis = 1000;

    private Type messageType;

    private MethodParameter methodParameter;

    private DefaultMQPushConsumer consumer;

    private MessageConverter messageConverter;

    private RocketMQTemplate rocketMqTemplate;

    public RocketMqListenerContainer() {
        log.info("RocketMqListenerContainer 初始化");
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        // 生成Consumer
        initRocketMqPushConsumer();

        this.messageType = getMessageType();
        this.methodParameter = getMethodParameter();
        log.debug("RocketMQ messageType: {}", messageType);
    }

    private Type getMessageType() {
        Type[] genericParameterTypes = method.getGenericParameterTypes();
        if (genericParameterTypes.length > 0) {
            return genericParameterTypes[0];
        }
        return Object.class;
    }

    private MethodParameter getMethodParameter() {
        return new MethodParameter(method, 0);
    }

    private void initRocketMqPushConsumer() throws MQClientException {
        if (bean == null || method == null) {
            throw new IllegalArgumentException("Property 'bean' and 'method' is required");
        }
        Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
        Assert.notNull(nameServer, "Property 'nameServer' is required");
        Assert.notNull(topic, "Property 'topic' is required");

        consumer = new DefaultMQPushConsumer(consumerGroup, true, StringUtils.EMPTY);
        consumer.setInstanceName(RocketMQUtil.getInstanceName(nameServer));
        consumer.setNamesrvAddr(nameServer);
        consumer.setConsumeThreadMin(consumeThreadMin);
        consumer.setConsumeThreadMax(consumeThreadMax);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.subscribe(topic, selectorExpression);
        // consumer获取到消息后，执行使用@MqConsumer的方法
        consumer.setMessageListener(new RocketMqMessageListenerConcurrently());
    }

    /**
     * 所有bean加载和初始化完毕执行，Spring容器回调该方法
     */
    @Override
    public void start() {
        if (this.isRunning()) {
            throw new IllegalStateException("container already running. " + this);
        }

        try {
            consumer.start();
        } catch (MQClientException e) {
            throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
        }
        this.setRunning(true);

        log.info("running container: {}", this);
    }

    @Override
    public void stop() {
        if (this.isRunning()) {
            if (Objects.nonNull(consumer)) {
                consumer.shutdown();
            }
            setRunning(false);
        }
    }

    public class RocketMqMessageListenerConcurrently implements MessageListenerConcurrently {

        /**
         * 消费消息
         * @param msgs 消息
         * @param context 上下文
         * @return 消费状态
         */
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            for (MessageExt messageExt : msgs) {
                log.info("received msg: {}", messageExt);
                try {
                    long now = System.currentTimeMillis();
                    handleMessage(messageExt);
                    long costTime = System.currentTimeMillis() - now;
                    log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
                } catch (Exception e) {
                    log.warn("consume message failed. messageId:{}, topic:{}, reconsumeTimes:{}", messageExt.getMsgId(), messageExt.getTopic(), messageExt.getReconsumeTimes(), e);
                    context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    }

    private void handleMessage(MessageExt messageExt) throws InvocationTargetException, IllegalAccessException {
        method.invoke(bean, doConvertMessage(messageExt));
        callback(messageExt);
    }

    /**
     * 消息消费后进行回调
     * @param messageExt 消息
     */
    private void callback(MessageExt messageExt) {
        // 获取消息属性主题名和tag名
        String callbackKey = messageExt.getProperties().get(RocketMqConstants.CALLBACK_KEY);
        String callbackTopicTag = messageExt.getProperties().get(RocketMqConstants.CALLBACK_TOPIC_TAG);
        if (StringUtils.isNoneBlank(callbackKey, callbackTopic)) {
            String topic = StringUtils.isBlank(callbackTopicTag) ? callbackTopic : callbackTopic + ":" + callbackTopicTag;
            log.info("回调主题：{}, key：{}", topic, callbackKey);
            Message<String> message = MessageBuilder.withPayload(callbackKey).build();
            rocketMqTemplate.syncSendOrderly(topic, message, callbackKey);
        }
    }

    private Object doConvertMessage(MessageExt messageExt) {
        if (Objects.equals(messageType, MessageExt.class)) {
            return messageExt;
        } else {
            String str = new String(messageExt.getBody(), Charset.forName(charset));
            if (Objects.equals(messageType, String.class)) {
                return str;
            } else {
                // If msgType not string, use objectMapper change it.
                try {
                    if (messageType instanceof Class) {
                        //if the messageType has not Generic Parameter
                        return this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) messageType);
                    } else {
                        //if the messageType has Generic Parameter, then use SmartMessageConverter#fromMessage with third parameter "conversionHint".
                        //we have validate the MessageConverter is SmartMessageConverter in this#getMethodParameter.
                        return ((SmartMessageConverter) this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) ((ParameterizedType) messageType).getRawType(), methodParameter);
                    }
                } catch (Exception e) {
                    log.info("convert failed. str:{}, msgType:{}", str, messageType);
                    throw new RuntimeException("cannot convert message to " + messageType, e);
                }
            }
        }
    }
}
